#include <cnotify.h>
#include <framework.h>
#include <modules.h>
#include <log.h>
#include <posix.h>
#include <template.h>
#include <taskManager.h>
#include <commands.h>

STATUS Cnotify::subRegReceiver(int recvId)
{

    std::lock_guard<std::mutex> lock{this->m_objectLock};
    auto iter = std::find(m_observerList.begin(), m_observerList.end(), recvId);
    COND_CHECK_RET_ERROR(iter != m_observerList.end());

    m_observerList.push_back(recvId);
    return OK;
}

STATUS Cnotify::subUnRegReceiver(int recvId)
{
    std::lock_guard<std::mutex> lock{this->m_objectLock};
    auto iter = std::find(m_observerList.begin(), m_observerList.end(), recvId);
    COND_CHECK_RET_ERROR(iter == m_observerList.end());

    m_observerList.erase(iter);
    return OK;
}

STATUS Cnotify::destory()
{
    stoped.store(true);
    cv_task.notify_all();
    return OK;
}

STATUS Cnotify::broadcast(int sendId, int cmdId, int recvfd, bool async)
{
    COND_CHECK_RET_ERROR(_UNLIKELY(this->stoped.load()));

    ArgList reqArgs(recvfd);
    if (async)
    {
        return asyncNotify(sendId, MODULE_ALL, cmdId, reqArgs);
    }
    else
    {
        TextTable outMsg;
        std::lock_guard<std::mutex> lock{this->m_objectLock};
        for (auto &iter : m_observerList)
        {
            (void)GET_FRAMEWORK_MGR->callObjectMethod<objbase>(
                iter,
                &objbase::process,
                static_cast<int>(sendId),
                static_cast<int>(iter),
                static_cast<int>(cmdId),
                reqArgs, outMsg);
        }
    }
    return OK;
}

STATUS Cnotify::notify(int sendId, int recvId, int cmdId, ArgList &reqArgs, TextTable &outMsg, bool async)
{
    COND_CHECK_RET_ERROR(_UNLIKELY(this->stoped.load()));

    if (_UNLIKELY(!async))
    {
        {
            std::lock_guard<std::mutex> lock{this->m_objectLock};
            auto iter = std::find(m_observerList.begin(), m_observerList.end(), recvId);
            COND_CHECK_RET_ERROR(iter == m_observerList.end());
        }

        return GET_FRAMEWORK_MGR->callObjectMethod<objbase, STATUS>(
            recvId,
            &objbase::process,
            static_cast<int>(sendId),
            static_cast<int>(recvId),
            static_cast<int>(cmdId),
            reqArgs, outMsg);
    }
    return asyncNotify(sendId, recvId, cmdId, reqArgs);
}

STATUS Cnotify::asyncNotify(int sendId, int recvId, int cmdId, ArgList &reqArgs)
{
    std::shared_ptr<AsyncArgc> pool = std::make_shared<AsyncArgc>(sendId, recvId, cmdId, reqArgs);
    std::lock_guard<std::mutex> lock{m_taskLock};
    tasks.emplace(pool);
    cv_task.notify_one();

    return OK;
}

STATUS Cnotify::asyncProc()
{
    while (!this->stoped)
    {
        std::shared_ptr<AsyncArgc> reqArgs;
        {
            std::unique_lock<std::mutex> lock{this->m_taskLock};

            this->cv_task.wait(lock,
                               [this]
                               {
                                   return this->stoped.load() ||
                                          !this->tasks.empty();
                               });
            if (this->stoped && this->tasks.empty())
            {
                break;
            }
            reqArgs = this->tasks.front();
            this->tasks.pop();
        }
        if (reqArgs != nullptr)
        {
            if (reqArgs->recvId == MODULE_ALL)
            {
                for (int recvId = MODULE_KERNEL; recvId < MODULE_BASE_END; recvId++)
                {
                    asyncNotify(reqArgs->sendId, recvId,
                                reqArgs->cmdId, reqArgs->reqMsg);
                }
            }
            else
            {
                TextTable outMsg;
                notify(reqArgs->sendId,
                       reqArgs->recvId,
                       reqArgs->cmdId,
                       reqArgs->reqMsg,
                       outMsg,
                       false);
            }
        }
    }
    return OK;
}
STATUS Cnotify::postInit()
{
    auto lambda = [](void* pobjPara) -> int {
        Cnotify* pThis = static_cast<Cnotify*>(pobjPara);
        pThis->asyncProc();
        return 0;
        };

    using FuncType = int (*)(void*);
    FuncType funcPtr = lambda;

    int idlThrMax = std::min(static_cast<int>(std::thread::hardware_concurrency()), 4);
    for (int idlThrNum = 0; idlThrNum < idlThrMax; idlThrNum++)
    {
       COND_CHECK_RET_ERROR(CREATE_OBJTASK(utils::format("NotifyTask%d", idlThrNum).c_str(),
           reinterpret_cast<FUNCPTR>(funcPtr),this) != OK);
    }
    
    COND_CHECK_RET_ERROR(subRegReceiver(MODULE_NOTIFY));

    return OK;
}
STATUS Cnotify::preInit()
{
    return OK;
}
STATUS Cnotify::dump(TextTable &t)
{
    t.setTableName("Cnotify");
    t.addHeader("objectId");
    t.addHeader("objectPtr").endOfRow();

    auto getObjectName = [](objbase* pObj, std::string & name)
        {
            name.clear();
            name = utils::to_hexstring(pObj);
        };
    std::lock_guard<std::mutex> lock{ this->m_objectLock };
    for (auto& iter : m_observerList)
    {
        std::string objectPtr;

        GET_FRAMEWORK_MGR->callObjectMethod<objbase>(iter, getObjectName, objectPtr);
        t.add(utils::to_hexstring(iter));
        t.add(objectPtr).endOfRow();        
    }
    return OK;
}

PROCESS_IMPL_DESC(Cnotify, process)
{
    PROCESS_BEGIN()
    PROCESS_CALL(CMD_SYS_SHUTDOWN, sysShutdown)
    PROCESS_END(OK)
}
PROCESS_IMPL_DESC(Cnotify, sysShutdown)
{
    return destory();
}
STATUS registerMessageListener(int recvId)
{
    return GET_FRAMEWORK_MGR->callObjectMethod<Cnotify, STATUS>
        (MODULE_NOTIFY, &Cnotify::subRegReceiver, static_cast<int>(recvId));
}
REG_TO_FRAMEWORK(Cnotify, MODULE_NOTIFY)